package day05;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.runtime.state.memory.MemoryStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class TestRecoeryListState {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setParallelism(2);
        //设置Checkpoint
        env.enableCheckpointing(1000);
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        env.getCheckpointConfig().setFailOnCheckpointingErrors(true);

        //设置外部存储
        // checkpointing每隔一段时间将状态保存到State Backend（默认异步）
        env.setStateBackend(new MemoryStateBackend(1024*1024*10,true));

        //设置恢复策略
        env.setRestartStrategy(
                RestartStrategies.fixedDelayRestart(
                        3, // number of restart attempts
                        Time.of(0, TimeUnit.SECONDS) // delay
                )
        );

        //listState的恢复
        DataStreamSource<String> ds = env.socketTextStream("192.168.21.160", 6667);
        ListStateDescriptor<String> listdesc = new ListStateDescriptor<>("listdesc", String.class);
        ds.map(new MyMapFunction()).print();

    }

    public static class MyMapFunction extends RichMapFunction<String, String> implements CheckpointedFunction {
        Integer index=null;
        List<String> list=new ArrayList<String>();
        ListState<String> state=null;

        @Override
        public void open(Configuration parameters) throws Exception {
            index=getRuntimeContext().getIndexOfThisSubtask();
        }

        @Override
        public String map(String value) throws Exception {
            if (value.equals("e")){
                int i=1/0;
            }
            list.add(index+"---->"+value);
            return value;
        }

        @Override
        public void snapshotState(FunctionSnapshotContext context) throws Exception {
            state.clear();
            state.addAll(list);
        }

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            ListStateDescriptor<String> listDesc = new ListStateDescriptor<>("listDesc", String.class);
            state=context.getOperatorStateStore().getUnionListState(listDesc);
            for(String s:state.get()){
                //System.out.println("index:"+s);
                System.out.println(getRuntimeContext().getIndexOfThisSubtask()+":"+s);
            }
        }
        //先恢复再运行

    }

}
